package cn.dmp.service

import java.util.Properties

import cn.dmp.beans.LogInfo
import cn.dmp.util.AppParams
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SQLContext, SaveMode}

/**
  * 3.2.1 地域分布
  */
object AreaDistribution {


  /**
    * 地域分布     方法一：使用spark-sql 的 sqlContext
    *
    * @param parquet
    * @param sqlContext
    */
  def getAeraDistribution(parquet: DataFrame, sqlContext: SQLContext,broadInfo: Broadcast[Map[String, String]] ): Unit = {

    // 28 ispname: String
    // 地域分布
    //                    int         int          int         int        int     int   int         Double      Double
    //provence    city    REQUESTMODE	PROCESSNODE	ISEFFECTIVE	ISBILLING	 ISBID	 ISWIN	ADORDERID  WinPrice    adpayment
    //24,          25     8,            35,           30,       31,       39,    42,     2          41           75
    //1            2      3             4              5         6         7      8      9          10           11

    /**
      * 维度统计
      * 24  provincename  String 省                                          1
      * 25  cityname String  市                                          2
      * 27 ispname: String, 运营商名称                                    12   3.2.2 运营商：
      * 29 networkmannername:String  联网方式名称                         13
      * 34 devicetype: Int, 设备类型（1：手机 2：平板  其他）                14
      * 17 client: Int, 操作系统 （1：android 2：ios 3：wp）               15
      * 14 appname: String, 应用名称                                      16   3.2.3
      * 55 storeurl: String, app 的市场下载地址                            17
      *
      *
      * 统计指标
      * 8 requestmode: Int, 数据请求方式（1:请求、2:展示、3:点击）             3
      * 35 processnode: Int, 流程节点（1：请求量 kpi 2：有效请求 3：广告请求）  4
      * 30 iseffective: Int, 有效标识（有效指可以正常计费的）(0：无效 1：有效    5
      * 31 isbilling: Int, 是否收费（0：未收费 1：已收费）                    6
      * 39 isbid: Int, 是否 rtb                                           7
      * 42 iswin: Int, 是否竞价成功                                        8
      * 2  adorderid: Int, 广告 id                                         9
      * 41 winprice: Double, rtb 竞价成功价格                              10
      * 75 adpayment: Double, 转换后的广告消费                              11
      *
      * "ispname","networkmannername","devicetype","client","appname","storeurl"
      */

    //val logRdd: RDD[Row] = parquet.rdd
    //val logInfo: RDD[LogInfo] = logRdd.map(row=>row.asInstanceOf[LogInfo])    可以通过对象获取信息

    //logRdd.foreach(println)
    // 还可以通过getAs读取
    /*val parquetInfo: RDD[(String, String, Int, Int, Int, Int, Int, Int, Int, Double,Double, String, String, Int, Int, String, String)] = parquet.map(line => {
      (
        line.getAs[String]("provincename"),
        line.getAs[String]("cityname"),
        line.getAs[Int]("requestmode"),
        line.getAs[Int]("processnode"),
        line.getAs[Int]("iseffective"),
        line.getAs[Int]("isbilling"),
        line.getAs[Int]("isbid"),
        line.getAs[Int]("iswin"),
        line.getAs[Int]("adorderid"),
        line.getAs[Double]("winprice"),
        line.getAs[Double]("adpayment"),
        line.getAs[String]("ispname"),
        line.getAs[String]("networkmannername"),
        line.getAs[Int]("devicetype"),
        line.getAs[Int]("client"),
        line.getAs[String]("appname"),
        line.getAs[String]("storeurl")
      )
    })*/

    //通过get读取
    val dataInfo: RDD[(String, String, Int, Int, Int, Int, Int, Int, Int, Double, Double, String, String, Int, Int, String, String,String)] = parquet.map(row => {
      (row.getString(24), row.getString(25), row.getInt(8), row.getInt(35), row.getInt(30),
        row.getInt(31), row.getInt(39), row.getInt(42), row.getInt(2), row.getDouble(41), row.getDouble(75),
        row.getString(27), row.getString(29), row.getInt(34), row.getInt(17), row.getString(14), row.getString(55),row.getString(13))
    })

    dataInfo.cache()

    //distributionInfo.foreach(println)
    import sqlContext.implicits._
    //按省分组
    val etlInfo: RDD[(String, String, Int, Int, Int, Int, Int, Int, Int, Double, Double, String, String, Int, Int, String, String)] = dataInfo.map(info => {
      val requestNum: Int = if (info._3 == 1 && info._4.toInt >= 1) 1 else 0
      val validNum: Int = if (info._3 == 1 && info._4.toInt >= 2) 1 else 0
      val adRequestNum: Int = if (info._3 == 1 && info._4 == 3) 1 else 0
      val participationNum: Int = if (info._5 == 1 && info._6 == 1 && info._7 == 1 && !(info._8 == 0)) 1 else 0
      val successfulsBidNum: Int = if (info._5 == 1 && info._6.toInt >= 1 && info._8 == 1) 1 else 0
      val showNum: Int = if (info._3 == 2 && info._5 == 1) 1 else 0
      val clickNum: Int = if (info._3 == 3 && info._5.toInt == 1) 1 else 0
      val dspConsume: Double = if (info._5 == 1 && info._6 == 1 && info._7 == 1) info._10.toDouble / 1000 else 0
      val dspCost: Double = if (info._5 == 1 && info._6 == 1 && info._7 == 1) info._11.toDouble / 1000 else 0
      val brocastMsg: Map[String, String] = broadInfo.value
      var appName=if("0".equals(info._16)||"".equals(info._16)) brocastMsg.getOrElse(info._18,info._18) else info._16

      (info._1, info._2, requestNum, validNum, adRequestNum, participationNum, successfulsBidNum, showNum, clickNum, dspConsume, dspCost,
        info._12, info._13, info._14, info._15, appName, info._17)
    })

    val etlDF: DataFrame = etlInfo.toDF("provence", "city", "requestNum", "validNum", "adRequestNum", "participationNum", "successfulsBidNum",
      "showNum", "clickNum", "dspConsume", "dspCost", "ispname", "networkmannername", "devicetype", "client", "appname", "storeurl")
    etlDF.cache()

    //3.2.1 地域分布
    etlDF.registerTempTable("t_provence_info")

    //按省分组
    sqlContext.sql(
      """
          select provence,sum(requestNum) requestNum,sum(validNum) validNum, sum(adRequestNum) adRequestNum,sum(participationNum) participationNum,
          sum(successfulsBidNum) successfulsBidNum,sum(showNum) showNum,sum(clickNum) clickNum,sum(dspConsume) dspConsume,sum(dspCost) dspCost
          from t_provence_info group by provence order by provence
      """.stripMargin).toDF.write.mode(SaveMode.Overwrite).jdbc(AppParams.url, AppParams.provence, AppParams.prop)


    //按省市分组
    sqlContext.sql(
      """
        select CONCAT(provence,":",city) provence,sum(requestNum) requestNum,sum(validNum) validNum, sum(adRequestNum) adRequestNum,sum(participationNum) participationNum,
        sum(successfulsBidNum) successfulsBidNum,sum(showNum) showNum,sum(clickNum) clickNum,sum(dspConsume) dspConsume,sum(dspCost) dspCost
        from t_provence_info group by provence,city order by provence,city
      """.stripMargin).toDF.write.mode(SaveMode.Append).jdbc(AppParams.url, AppParams.provence, AppParams.prop)


    //3.2.2 运营商：27 ispname: String, 运营商名称
    etlDF.registerTempTable("t_ispname_info")
    sqlContext.sql(
      """
        |select ispname,count(1) totalRequest,sum(requestNum) requestNum,sum(validNum) validNum, sum(adRequestNum) adRequestNum,sum(participationNum) participationNum,
        |        sum(successfulsBidNum) successfulsBidNum,sum(showNum) showNum,sum(clickNum) clickNum,sum(dspConsume) dspConsume,sum(dspCost) dspCost
        |        from t_ispname_info group by ispname order by ispname
      """.stripMargin).toDF.write.mode(SaveMode.Overwrite).jdbc(AppParams.url, AppParams.ispname, AppParams.prop)


    //3.2.2 网络类型：29 networkmannername:String  联网方式名称
    etlDF.registerTempTable("t_networkmannername_info")
    sqlContext.sql(
      """
        |select networkmannername,count(1) totalRequest,sum(requestNum) requestNum,sum(validNum) validNum, sum(adRequestNum) adRequestNum,sum(participationNum) participationNum,
        |        sum(successfulsBidNum) successfulsBidNum,sum(showNum) showNum,sum(clickNum) clickNum,sum(dspConsume) dspConsume,sum(dspCost) dspCost
        |        from t_networkmannername_info group by networkmannername order by networkmannername
      """.stripMargin).toDF.write.mode(SaveMode.Overwrite).jdbc(AppParams.url, AppParams.networkmannername, AppParams.prop)

    //3.2.2 设备类型 34 devicetype: Int, 设备类型（1：手机 2：平板  其他）
    etlDF.registerTempTable("t_devicetype_info")
    sqlContext.sql(
      """
        |select
        |case
        |when networkmannername="1" then "手机"
        |when networkmannername="2" then "平板"
        |else "其他"
        |end
        |as networkmannername,
        |count(1) totalRequest,sum(requestNum) requestNum,sum(validNum) validNum, sum(adRequestNum) adRequestNum,sum(participationNum) participationNum,
        |        sum(successfulsBidNum) successfulsBidNum,sum(showNum) showNum,sum(clickNum) clickNum,sum(dspConsume) dspConsume,sum(dspCost) dspCost
        |        from t_devicetype_info group by networkmannername order by networkmannername
      """.stripMargin).toDF.write.mode(SaveMode.Overwrite).jdbc(AppParams.url, AppParams.devicetype, AppParams.prop)


    //3.2.2 操作系统： client: Int, 操作系统 （1：android 2：ios 3：wp）
    etlDF.registerTempTable("t_client_info")
    sqlContext.sql(
      """
        |select
        |case
        |when client="1" then "android"
        |when client="2" then "ios"
        |when client="3" then "wp"
        |else "其他"
        |end
        |as client,count(1) totalRequest,sum(requestNum) requestNum,sum(validNum) validNum, sum(adRequestNum) adRequestNum,sum(participationNum) participationNum,
        |        sum(successfulsBidNum) successfulsBidNum,sum(showNum) showNum,sum(clickNum) clickNum,sum(dspConsume) dspConsume,sum(dspCost) dspCost
        |        from t_client_info group by client order by client
      """.stripMargin).toDF.write.mode(SaveMode.Overwrite).jdbc(AppParams.url, AppParams.client, AppParams.prop)


    //3.2.3 媒体分析   appname: String, 应用名称
    etlDF.registerTempTable("t_appname_info")
    sqlContext.sql(
      """
        |select appname,count(1) totalRequest,sum(requestNum) requestNum,sum(validNum) validNum, sum(adRequestNum) adRequestNum,sum(participationNum) participationNum,
        |        sum(successfulsBidNum) successfulsBidNum,sum(showNum) showNum,sum(clickNum) clickNum,sum(dspConsume) dspConsume,sum(dspCost) dspCost
        |        from t_appname_info group by appname order by appname
      """.stripMargin).toDF.write.mode(SaveMode.Overwrite).jdbc(AppParams.url, AppParams.appname, AppParams.prop)


    //3.2.4 渠道报表   storeurl: String, app 的市场下载地址
    etlDF.registerTempTable("t_storeurl_info")
    sqlContext.sql(
      """
        |select storeurl,count(1) totalRequest,sum(requestNum) requestNum,sum(validNum) validNum, sum(adRequestNum) adRequestNum,sum(participationNum) participationNum,
        |        sum(successfulsBidNum) successfulsBidNum,sum(showNum) showNum,sum(clickNum) clickNum,sum(dspConsume) dspConsume,sum(dspCost) dspCost
        |        from t_storeurl_info group by storeurl order by storeurl
      """.stripMargin).toDF.write.mode(SaveMode.Overwrite).jdbc(AppParams.url, AppParams.storeurl, AppParams.prop)


  }


  /**
    * 地域分布 ---方法二   使用spark-core 算子reduceByKey
    *
    * @param parquet
    * @param sqlContext
    */
  def getAeraDistributionV2(parquet: DataFrame, sqlContext: SQLContext): Unit = {

    /**
      * 维度统计
      * 24  provence  String 省                                          1
      * 25  provence String  市                                          2
      * 27 ispname: String, 运营商名称                                    12   3.2.2 运营商：
      * 29 networkmannername:String  联网方式名称                         13
      * 34 devicetype: Int, 设备类型（1：手机 2：平板  其他）                14
      * 17 client: Int, 操作系统 （1：android 2：ios 3：wp）               15
      * 14 appname: String, 应用名称                                      16   3.2.3
      * 55 storeurl: String, app 的市场下载地址                            17
      *
      *
      * 统计指标
      * 8 requestmode: Int, 数据请求方式（1:请求、2:展示、3:点击）             3
      * 35 processnode: Int, 流程节点（1：请求量 kpi 2：有效请求 3：广告请求）  4
      * 30 iseffective: Int, 有效标识（有效指可以正常计费的）(0：无效 1：有效    5
      * 31 isbilling: Int, 是否收费（0：未收费 1：已收费）                    6
      * 39 isbid: Int, 是否 rtb                                           7
      * 42 iswin: Int, 是否竞价成功                                        8
      * 2 adorderid: Int, 广告 id                                         9
      * 41 winprice: Double, rtb 竞价成功价格                              10
      * 75 adpayment: Double, 转换后的广告消费                              11
      *
      * "ispname","networkmannername","devicetype","client","appname","storeurl"
      */

    // 地域分布
    val logRdd: RDD[Row] = parquet.rdd
    //val logInfo: RDD[LogInfo] = logRdd.map(row=>row.asInstanceOf[LogInfo])    可以通过对象获取信息

    //logRdd.foreach(println)
    val dataInfo: RDD[(String, String, Int, Int, Int, Int, Int, Int, Int, Double, Double)] = logRdd.map(row => {
      (row.getString(24), row.getString(25), row.getInt(8), row.getInt(35), row.getInt(30),
        row.getInt(31), row.getInt(39), row.getInt(42), row.getInt(2), row.getDouble(41), row.getDouble(75))
    })

    dataInfo.cache()
    import sqlContext.implicits._

    /*val dataInfoRes: RDD[(String, String, Int, Int, Int, Int, Int, Int, Int, Double, Double, String, String, Int, Int, String, String)] = logRdd.map(row => {
      (row.getString(24), row.getString(25), row.getInt(8), row.getInt(35), row.getInt(30),
        row.getInt(31), row.getInt(39), row.getInt(42), row.getInt(2), row.getDouble(41), row.getDouble(75),
        row.getString(27), row.getString(29), row.getInt(34), row.getInt(17), row.getString(14), row.getString(55))
    })

    //distributionInfo.foreach(println)
    import sqlContext.implicits._
    //按省分组
    val etlData: RDD[((String, String, String, String, Int, Int, String, String), List[Double])] = dataInfoRes.map(info => {
      val requestNum: Int = if (info._3 == 1 && info._4.toInt >= 1) 1 else 0
      val validNum: Int = if (info._3 == 1 && info._4.toInt >= 2) 1 else 0
      val adRequestNum: Int = if (info._3 == 1 && info._4 == 3) 1 else 0
      val participationNum: Int = if (info._5 == 1 && info._6 == 1 && info._7 == 1 && !(info._8 == 0)) 1 else 0
      val successfulsBidNum: Int = if (info._5 == 1 && info._6.toInt >= 1 && info._8 == 1) 1 else 0
      val showNum: Int = if (info._3 == 2 && info._5 == 1) 1 else 0
      val clickNum: Int = if (info._3 == 3 && info._5.toInt == 1) 1 else 0
      val dspConsume: Double = if (info._5 == 1 && info._6 == 1 && info._7 == 1) info._10.toDouble / 1000 else 0
      val dspCost: Double = if (info._5 == 1 && info._6 == 1 && info._7 == 1) info._11.toDouble / 1000 else 0
      ((info._1, info._2, info._12, info._13, info._14, info._15, info._16, info._17), List[Double](requestNum, validNum, adRequestNum, participationNum, successfulsBidNum, showNum, clickNum, dspConsume, dspCost))
    })

    //
    val groupInfo: RDD[((String, String), Iterable[((String, String, String, String, Int, Int, String, String), List[Double])])] = etlData.groupBy(t=>(t._1._1,t._1._2))

    val provenceInfo: RDD[((String, String), List[Double])] = groupInfo.map(t => {
      //先group后reduce    ——》  取代reduceByKey
      val iterinfo: Iterable[List[Double]] = t._2.map(itr => itr._2)
      val result: List[Double] = iterinfo.reduce((list1, list2) => {
        list1.zip(list2).map(info => info._1 + info._2)
      })
      (t._1, result)
    })*/
    //provenceInfo.foreach(println)

    //按省分组
    val etlInfo: RDD[(String, List[Double])] = dataInfo.map(info => {
      val requestNum: Int = if (info._3 == 1 && info._4.toInt >= 1) 1 else 0
      val validNum: Int = if (info._3 == 1 && info._4.toInt >= 2) 1 else 0
      val adRequestNum: Int = if (info._3 == 1 && info._4 == 3) 1 else 0
      val participationNum: Int = if (info._5 == 1 && info._6 == 1 && info._7 == 1 && !(info._8 == 0)) 1 else 0
      val successfulsBidNum: Int = if (info._5 == 1 && info._6.toInt >= 1 && info._8 == 1) 1 else 0
      val showNum: Int = if (info._3 == 2 && info._5 == 1) 1 else 0
      val clickNum: Int = if (info._3 == 3 && info._5.toInt == 1) 1 else 0
      val dspConsume: Double = if (info._5 == 1 && info._6 == 1 && info._7 == 1) info._10.toDouble / 1000 else 0
      val dspCost: Double = if (info._5 == 1 && info._6 == 1 && info._7 == 1) info._11.toDouble / 1000 else 0
      (info._1, List[Double](requestNum, validNum, adRequestNum, participationNum, successfulsBidNum, showNum, clickNum, dspConsume, dspCost))
    })

    val resultPro: RDD[(String, List[Double])] = etlInfo.reduceByKey((list1, list2) => {
      list1.zip(list2).map(t => t._1 + t._2)
    })
    val results: DataFrame = resultPro.map(tp => {
      (tp._1, tp._2(0), tp._2(1), tp._2(2), tp._2(3), tp._2(4), tp._2(5), tp._2(6), tp._2(7), tp._2(8))
    }).toDF("provence","requestNum", "validNum", "adRequestNum", "participationNum", "successfulsBidNum", "showNum", "clickNum", "dspConsume", "dspCost")
    //resultPro.foreach(println)

    val prop: Properties = new Properties()
    prop.setProperty("user", AppParams.user)
    prop.setProperty("password", AppParams.password)
    prop.setProperty("driver", AppParams.driver)

    results.write.mode(SaveMode.Overwrite).jdbc(AppParams.url, AppParams.provenceV1, prop)

    /*//按省市分组
    val etlInfoCity: RDD[((String, String), List[Double])] = dataInfo.map(info => {
      val requestNum: Int = if (info._3 == 1 && info._4.toInt >= 1) 1 else 0
      val validNum: Int = if (info._3 == 1 && info._4.toInt >= 2) 1 else 0
      val adRequestNum: Int = if (info._3 == 1 && info._4 == 3) 1 else 0
      val participationNum: Int = if (info._5 == 1 && info._6 == 1 && info._7 == 1 && !(info._8 == 0)) 1 else 0
      val successfulsBidNum: Int = if (info._5 == 1 && info._6.toInt >= 1 && info._8 == 1) 1 else 0
      val showNum: Int = if (info._3 == 2 && info._5 == 1) 1 else 0
      val clickNum: Int = if (info._3 == 3 && info._5.toInt == 1) 1 else 0
      val dspConsume: Double = if (info._5 == 1 && info._6 == 1 && info._7 == 1) info._10.toDouble / 1000 else 0
      val dspCost: Double = if (info._5 == 1 && info._6 == 1 && info._7 == 1) info._11.toDouble / 1000 else 0
      ((info._1, info._2), List[Double](requestNum, validNum, adRequestNum, participationNum, successfulsBidNum, showNum, clickNum, dspConsume, dspCost))
    })

    val resultProCity: RDD[((String, String), List[Double])] = etlInfoCity.reduceByKey((list1, list2) => {
      list1.zip(list2).map(t => t._1 + t._2)
    })
    resultProCity.map(info => {
      (info._1._1 + ":" + info._1._2, info._2)
    }).toDF.write.mode(SaveMode.Append).jdbc(AppParams.url, AppParams.provence, prop)*/
  }

}
